package com.zdb.demo.flink.streaming;

import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Iterator;

/**
 * 请实现一个 Flink Stream Job，实现如下功能
 *
 *
 * 使用 KeyedState（不使用 TimeAssigner）实现一个简单的 1 分钟的窗口（相当于 tumbling time window）
 * 时间从数据中抽取
 * 可以假设数据无乱序，所有数据的时间单调递增
 *
 */
public class H6 {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration(){{
            setInteger("rest.port", 9191);
            setBoolean("local.start-webserver", true);
        }};
        final StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        streamEnv.setParallelism(4);
        streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        SingleOutputStreamOperator<Tuple3<String, Long, Long>> ds = streamEnv.socketTextStream("localhost", 9999).filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                return value!=null && value.length() > 0;
            }
        }).map(new MapFunction<String, Tuple3<String, Long, Long>>() {
            @Override
            public Tuple3<String, Long, Long> map(String value) throws Exception {
                String[] strs = value.split(",");
                return new Tuple3<String, Long, Long>(strs[0], Long.parseLong(strs[1].trim()), Long.parseLong(strs[2].trim()) * 1000);
            }
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, Long, Long>>(Time.seconds(2)) {
            @Override
            public long extractTimestamp(Tuple3<String, Long, Long> element) {
                return element.f2;
            }
        }).keyBy(0)
                .flatMap(new RichFlatMapFunction<Tuple3<String, Long, Long>, Tuple3<String, Long, Long>>() {

                    private ListState<Tuple3<String, Long, Long>> listState;
                    private ValueState<Long> startTimeState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ListStateDescriptor<Tuple3<String, Long, Long>> descriptor = new ListStateDescriptor<Tuple3<String, Long, Long>>("list",
                                TypeInformation.of(new TypeHint<Tuple3<String, Long, Long>>() {}));
                        listState = getRuntimeContext().getListState(descriptor);

                        ValueStateDescriptor<Long> valueStateDescriptor = new ValueStateDescriptor<Long>("", Long.class);
                        startTimeState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void flatMap(Tuple3<String, Long, Long> value, Collector<Tuple3<String, Long, Long>> out) throws Exception {
//                        Iterator<Tuple3<String, Long, Long>> iter = listState.get().iterator();
                        // 如果当前值减最小值 > 60 * 1000 则输出liststate的值，并清空state
                        if(startTimeState.value() == null || startTimeState.value() == 0) { // 第一个
                            startTimeState.update(value.f2);
                            listState.add(value);
                        } else {
                            if(value.f2 - startTimeState.value() > 60000) {
                                Iterator<Tuple3<String, Long, Long>> iter = listState.get().iterator();
                                while (iter.hasNext()) {
                                    // 输出
                                    out.collect(iter.next());
                                }
                                listState.clear();
                                startTimeState.update(value.f2);
                                listState.add(value);
                            } else {
                                // 没有到时间，暂存起来
                                listState.add(value);
                            }
                        }
                    }
                });

        ds.print("window 1 minute------>");

        streamEnv.execute("SideOutputDemo stream job");
    }

}
